package com.zz.study.sharding.task;

import cn.hutool.core.date.DateUnit;
import com.zz.study.sharding.algorithm.compress.DPUtil;
import com.zz.study.sharding.component.StarRocksStreamWriter;
import com.zz.study.sharding.constants.Constants;
import com.zz.study.sharding.entity.DateSplit;
import com.zz.study.sharding.entity.DtFusdataFlink;
import com.zz.study.sharding.entity.DtFusdataFlinkSimple;
import com.zz.study.sharding.mappers.CompressTrackDataMapper;
import com.zz.study.sharding.mappers.SlaveDtFusdataFlinkMapper;
import com.zz.study.sharding.mappers.TrackdataHisTestMapper;
import com.zz.study.sharding.util.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.sql.Timestamp;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Future;

import static com.zz.study.sharding.constants.Constants.*;
import static com.zz.study.sharding.util.DateSplitUtils.splitByDay;
import static com.zz.study.sharding.util.DateUtil.DEFAULT_FORMAT;

/**
 * created by xjx on  2024-01-25 9:27
 */
@Component
@Slf4j
@EnableScheduling
public class TrackDataCompressTask {

    @Autowired
    private TrackdataHisTestMapper trackdataHisTestMapper;

    @Autowired
    private SlaveDtFusdataFlinkMapper slaveDtFusdataFlinkMapper;

    @Autowired
    private CompressTrackDataMapper compressTrackDataMapper;

    @Value("${algorithms.compress.dsMax}")
    private double dsMax;

    @Autowired
    private StarRocksStreamWriter starRocksStreamWriter;

    @Resource(name="shipTrackQueryExecutor")
    private ThreadPoolTaskExecutor shipTrackQueryExecutor;

    @Value("${batchWriteSize}")
    private int batchWriteSize;

    @Value("${compressBatchHours}")
    private int compressBatchHours;

//    public void scheduleCompressTask(){
//        Date now = new Date();
//        Date before = DateUtil.getAddDayDate(now,-1);
//        Date startTime = DateUtil.getDayBegin(before);
//        Date endTime = DateUtil.getDayEnd(before);
//        String targetTableName = DateUtil.getShardTableName("trackdata_his_",before);
//        List<String> targetIdList =  slaveDtFusdataFlinkMapper.distinctTarIDByTimeArea(targetTableName,startTime,endTime);
//        int batchSize=20;
//        if(!CollectionUtils.isEmpty(targetIdList)){
//            int batchCount = MathUtil.ceil(targetIdList.size(),batchSize);
//
//            for (int i = 0; i < batchCount; i++) {
//                if (i == batchCount-1) {
//                    List<String> targetIdListTemp= targetIdList.subList(batchSize * i, targetIdList.size());
//                    if(!CollectionUtils.isEmpty(targetIdListTemp)){
//                        batchInsertRecord(targetIdListTemp,startTime,endTime);
//                    }
//
//                } else {
//                    List<String> tarIdListTemp= targetIdList.subList(batchSize * i, batchSize*(i+1));
//                    if(!CollectionUtils.isEmpty(tarIdListTemp)){
//                        batchInsertRecord(tarIdListTemp,startTime,endTime);
//                    }
//                }
//            }
//        }
//    }
    @Scheduled(cron = "0 10/30 * * * ?")
    //@Scheduled(cron = "0/30 * * * * ?")
    public synchronized void scheduleCompressTrackdata(){
        Date now = new Date();
        String targetTableName ="";
         targetTableName = DateUtil.getShardTableName(CompressHisTrackDataTableNamePrefix,now);
        Long id  = compressTrackDataMapper.getMaxRecordId(targetTableName);
        if(Objects.isNull(id)||id<=0){
            Date fiveDayBefore  = cn.hutool.core.date.DateUtil.offsetDay(now,-15);
            targetTableName = DateUtil.getShardTableName(CompressHisTrackDataTableNamePrefix,fiveDayBefore);
             id  = compressTrackDataMapper.getMaxRecordId(targetTableName);
        }
        List<DtFusdataFlinkSimple> maxRecordList = compressTrackDataMapper.getRecordByID(targetTableName,id);
        DtFusdataFlinkSimple maxRecord = new DtFusdataFlinkSimple();
        if(!CollectionUtils.isEmpty(maxRecordList)){
            maxRecord = maxRecordList.get(0);
        }
        Date maxRecordTime = maxRecord.getRecordTime();
        Date expectEndTime = cn.hutool.core.date.DateUtil.offsetHour(maxRecordTime,compressBatchHours);

        Date startTime = maxRecordTime;
        Date endTime = expectEndTime;
        while(!historyDataExist(startTime,endTime)){
            startTime = cn.hutool.core.date.DateUtil.offsetHour(startTime,compressBatchHours);
            endTime = cn.hutool.core.date.DateUtil.offsetHour(endTime,compressBatchHours);
        }
        log.error("压缩数据开始========================startTime=="+ cn.hutool.core.date.DateUtil.format(startTime,"yyyy-MM-dd HH:mm:ss")+
                "endTime ====="+cn.hutool.core.date.DateUtil.format(endTime,"yyyy-MM-dd HH:mm:ss"));
        long start = System.currentTimeMillis();
        writeCompressDataByDateRange(startTime,endTime);
        log.error("压缩数据完成=========================耗时"+(System.currentTimeMillis()-start)+"");
    }


    public boolean historyDataExist(Date startTime,Date endTime){
        String targetTableName = DateUtil.getShardTableName(HisTrackDataTableNamePrefix,endTime);
        List<String> targetIdList =  slaveDtFusdataFlinkMapper.distinctTarIDByTimeArea(targetTableName,startTime,endTime);
        if(CollectionUtils.isEmpty(targetIdList)){
            return false;
        }
        return true;
    }

    public void writeCompressDataByDateRange(Date startTime,Date endTime){
        log.error("开始压缩数据=====startTime=="+ cn.hutool.core.date.DateUtil.format(startTime,"yyyy-MM-dd HH:mm:ss")+
                "endTime ====="+cn.hutool.core.date.DateUtil.format(endTime,"yyyy-MM-dd HH:mm:ss"));
        String targetTableName = DateUtil.getShardTableName(HisTrackDataTableNamePrefix,endTime);
        List<String> targetIdList =  slaveDtFusdataFlinkMapper.distinctTarIDByTimeArea(targetTableName,startTime,endTime);
        log.error("targetIdList数组长度====="+targetIdList.size()+"targetTableName==="+targetTableName);
        int batchSize=500;
        if(!CollectionUtils.isEmpty(targetIdList)){
            int batchCount = MathUtil.ceil(targetIdList.size(),batchSize);
            for (int j = 0; j < batchCount; j++) {
                if (j == batchCount-1) {
                    List<String> targetIdListTemp= targetIdList.subList(batchSize * j, targetIdList.size());
                    if(!CollectionUtils.isEmpty(targetIdListTemp)){
                        log.error("开始导入第"+j+"批次的数据=====当天共有"+batchCount+"批次");
                        long start = System.currentTimeMillis();
                        //ConcurrentUtil.doJob(shipTrackQueryExecutor, new TrackQueryTask(slaveDtFusdataFlinkMapper,targetIdListTemp,startTime,endTime,dsMax,starRocksStreamWriter));
                        compressTrackData(targetIdListTemp,startTime,endTime);
                        log.error("导入完成第"+j+"批次的数据=====当天共有"+batchCount+"批次耗时"+(System.currentTimeMillis()-start)+"ms");
                    }
                } else {
                    List<String> tarIdListTemp= targetIdList.subList(batchSize * j, batchSize*(j+1));
                    if(!CollectionUtils.isEmpty(tarIdListTemp)){
                        log.error("开始导入第"+j+"批次的数据=====当天共有"+batchCount+"批次");
                        long start = System.currentTimeMillis();
                        //ConcurrentUtil.doJob(shipTrackQueryExecutor, new TrackQueryTask(slaveDtFusdataFlinkMapper,tarIdListTemp,startTime,endTime,dsMax,starRocksStreamWriter));
                        compressTrackData(tarIdListTemp,startTime,endTime);
                        log.error("导入完成第"+j+"批次的数据=====当天共有"+batchCount+"批次耗时"+(System.currentTimeMillis()-start)+"ms");
                    }
                }
            }
        }



    }

    public void writeCompressDataByDateRange(String startTimeStr,String endTimeStr){
            long start = System.currentTimeMillis();
            try{
                Date startTime = DateUtil.formatTime(startTimeStr,DEFAULT_FORMAT);
                Date endTime = DateUtil.formatTime(endTimeStr,DEFAULT_FORMAT);
                long betweenDay = cn.hutool.core.date.DateUtil.between(startTime, endTime, DateUnit.DAY);
                if(betweenDay>1){
                    List<DateSplit> dateSplitList = DateSplitUtils.splitByHour(startTime,endTime,12);
                    for(int i=0;i<dateSplitList.size();i++){
                        try {
                        log.error("时间大于1天，分批次压缩，第"+i+"批次=========="+"开始时间====="+dateSplitList.get(i).getStartDateTimeStr() +"结束时间===="+dateSplitList.get(i).getEndDateTimeStr());
                        Date startTime2 = dateSplitList.get(i).getStartDateTime();
                        Date endTime2 = dateSplitList.get(i).getEndDateTime();
                        writeCompressDataByDateRange(startTime2,endTime2);
                            //导入1天数据之后，得休息5分钟，导入太多，数据库会崩掉
                            Thread.sleep(1000*60*5);
                        } catch (Exception e) {
                            log.error("写入数据报错======"+e.getMessage());
                            Thread.sleep(1000*60*5);
                            e.printStackTrace();
                        }
                    }
                }else{
                    writeCompressDataByDateRange(startTime,endTime);
                }
            }catch (Exception e){
                log.error(e.getMessage());
            }
            log.error("导入数据完成耗时"+(System.currentTimeMillis()-start)+"ms");

    }



    public void lastWeekTrackDataCompress(){

        for(int i=1;i<3;i++){
            Date now = new Date();
            Date before = DateUtil.getAddDayDate(now,-i);
            Date startTime = DateUtil.getDayBegin(before);
            Date endTime = DateUtil.getDayEnd(before);
            String targetTableName = DateUtil.getShardTableName(HisTrackDataTableNamePrefix,before);
            List<String> targetIdList =  slaveDtFusdataFlinkMapper.distinctTarIDByTimeArea(targetTableName,startTime,endTime);
            log.error("targetIdList数组长度====="+targetIdList.size());
            int batchSize=1000;
            if(!CollectionUtils.isEmpty(targetIdList)){
                int batchCount = MathUtil.ceil(targetIdList.size(),batchSize);
                for (int j = 0; j < batchCount; j++) {
                    if (j == batchCount-1) {
                        List<String> targetIdListTemp= targetIdList.subList(batchSize * j, targetIdList.size());
                        if(!CollectionUtils.isEmpty(targetIdListTemp)){
                            log.error("开始导入第"+i+"天第"+j+"批次的数据=====当天共有"+batchCount+"批次");
                            //ConcurrentUtil.doJob(shipTrackQueryExecutor, new TrackQueryTask(slaveDtFusdataFlinkMapper,targetIdListTemp,startTime,endTime,dsMax,starRocksStreamWriter));
                            compressTrackData(targetIdListTemp,startTime,endTime);
                            log.error("导入完成第"+i+"天第"+j+"批次的数据=====当天共有"+batchCount+"批次");
                        }
                    } else {
                        List<String> tarIdListTemp= targetIdList.subList(batchSize * j, batchSize*(j+1));
                        if(!CollectionUtils.isEmpty(tarIdListTemp)){
                            log.error("开始导入第"+i+"天第"+j+"批次的数据=====当天共有"+batchCount+"批次");
                            //ConcurrentUtil.doJob(shipTrackQueryExecutor, new TrackQueryTask(slaveDtFusdataFlinkMapper,tarIdListTemp,startTime,endTime,dsMax,starRocksStreamWriter));
                            compressTrackData(tarIdListTemp,startTime,endTime);
                            log.error("导入完成第"+i+"天第"+j+"批次的数据=====当天共有"+batchCount+"批次");
                        }
                    }
                }
            }
        }


    }


    /**
     *
     * @param tarIdList  1000到2000个tarID
     * @param startTime
     * @param endTime
     * @return
     */
    public void compressTrackData(List<String> tarIdList,Date startTime,Date endTime)   {

        String targetCompressDataTableName = DateUtil.getShardTableName(CompressHisTrackDataTableNamePrefix,startTime);

        int batchFetchSize=20;
        int batchCount = MathUtil.ceil(tarIdList.size(),batchFetchSize);
        List<Future<List<Map<String,Object>>>> futureList = new CopyOnWriteArrayList<>();
        List<Map<String,Object>> afterCompressDataList = new ArrayList<>();
        for (int j = 0; j < batchCount; j++) {
            if (j == batchCount-1) {
                List<String> targetIdListTemp= tarIdList.subList(batchFetchSize * j, tarIdList.size());
                if(!CollectionUtils.isEmpty(targetIdListTemp)){
                    Future<List<Map<String,Object>>> compressTrackdataListFuture =  ConcurrentUtil.doJob(shipTrackQueryExecutor, new TrackQueryTask(slaveDtFusdataFlinkMapper,targetIdListTemp,startTime,endTime,dsMax,starRocksStreamWriter));
                    if(!CollectionUtils.isEmpty((List<Map<String,Object>>) ConcurrentUtil.futureGet(compressTrackdataListFuture))){
                        futureList.add(compressTrackdataListFuture);
                    }

                }
            } else {
                List<String> tarIdListTemp= tarIdList.subList(batchFetchSize * j, batchFetchSize*(j+1));
                if(!CollectionUtils.isEmpty(tarIdListTemp)){
                    Future<List<Map<String,Object>>> compressTrackdataListFuture =  ConcurrentUtil.doJob(shipTrackQueryExecutor, new TrackQueryTask(slaveDtFusdataFlinkMapper,tarIdListTemp,startTime,endTime,dsMax,starRocksStreamWriter));
                    if(!CollectionUtils.isEmpty((List<Map<String,Object>>) ConcurrentUtil.futureGet(compressTrackdataListFuture))){
                        futureList.add(compressTrackdataListFuture);
                    }

                }
            }
        }
        if(!futureList.isEmpty()){
            for(Future future:futureList){
                List<Map<String,Object>> trackDataList = (List<Map<String,Object>>) ConcurrentUtil.futureGet(future);
                if(!CollectionUtils.isEmpty(trackDataList)){
                    afterCompressDataList.addAll(trackDataList);
                }
            }
        }else{
            log.error("没有数据====");
        }

        log.error("经过并发获取压缩之后，afterCompressDataList长度是===="+afterCompressDataList.size());
        List<Map<String,Object>> afterDistinctList = BeanUtil.distinctByKeyField(afterCompressDataList,"id");
        log.error("经过去重之后,afterDistinctList===="+afterDistinctList.size());
        int batchCount2 = MathUtil.ceil(afterDistinctList.size(),batchWriteSize);
        log.error("batchCount2===="+batchCount2);
        for (int j = 0; j < batchCount2; j++) {
            if (j == batchCount2-1) {
                List<Map<String,Object>>  beanListTemp = afterDistinctList.subList(batchWriteSize * j, afterDistinctList.size());
                if(!CollectionUtils.isEmpty(beanListTemp)){
                    //第一个元素
                    Map<String,Object> bean0=   beanListTemp.get(0);
                    String dateTimeStr0 ="";
                    if(bean0.get("RecordTime") instanceof Timestamp){
                        Date recordTime = DateUtil.timestampToDate((Timestamp) bean0.get("RecordTime"));
                        dateTimeStr0 = DateUtil.getTime(recordTime);
                    }else if(bean0.get("RecordTime") instanceof String){
                        dateTimeStr0 = (String)bean0.get("RecordTime");
                    }
                    String targetTableName0 = DateUtil.getShardTableName(CompressHisTrackDataTableNamePrefix,dateTimeStr0);
                    //最后一个元素
                    Map<String,Object> bean1=   beanListTemp.get(beanListTemp.size()-1);
                    String dateTimeStr1 ="";
                    if(bean1.get("RecordTime") instanceof Timestamp){
                        Date recordTime = DateUtil.timestampToDate((Timestamp) bean1.get("RecordTime"));
                        dateTimeStr1 = DateUtil.getTime(recordTime);
                    }else if(bean1.get("RecordTime") instanceof String){
                        dateTimeStr1 = (String)bean1.get("RecordTime");
                    }
                    String targetTableName1 = DateUtil.getShardTableName(CompressHisTrackDataTableNamePrefix,dateTimeStr1);
                    if(targetTableName0.equals(targetTableName1)){
                        write2StarRocks(beanListTemp,targetTableName1);
                    }else{
                        List<Map<String,Object>> beanList1 = new ArrayList<>();
                        List<Map<String,Object>> beanList2 = new ArrayList<>();
                        String targetTableNameTemp = "";
                        for(Map<String,Object> tempBean:beanListTemp){
                            String dateTimeStrTemp ="";
                            if(tempBean.get("RecordTime") instanceof Timestamp){
                                Date recordTime = DateUtil.timestampToDate((Timestamp) tempBean.get("RecordTime"));
                                dateTimeStrTemp = DateUtil.getTime(recordTime);
                            }else if(bean0.get("RecordTime") instanceof String){
                                dateTimeStrTemp = (String)tempBean.get("RecordTime");
                            }
                             targetTableNameTemp = DateUtil.getShardTableName(CompressHisTrackDataTableNamePrefix,dateTimeStrTemp);
                            if(dateTimeStr0.equals(targetTableNameTemp)){
                                beanList1.add(tempBean);
                            }else{
                                beanList2.add(tempBean);
                            }
                        }
                        write2StarRocks(beanList1,targetTableName0);
                        write2StarRocks(beanList2,targetTableName1);
                    }
                }
            } else {
                List<Map<String,Object>>  beanListTemp = afterDistinctList.subList(batchWriteSize * j, batchWriteSize*(j+1));
                if(!CollectionUtils.isEmpty(beanListTemp)){
                    write2StarRocks(beanListTemp,targetCompressDataTableName);
                }
            }
        }
    }


    public void write2StarRocks(List<Map<String,Object>>  beanList,String targetTable){
        try{
            String flushData = JsonUtil.serialize(beanList);
            if(StringUtils.isNotEmpty(flushData)){
                starRocksStreamWriter.importBeanMapList2Doris(beanList, targetTable, Constants.trackDataColumns);
                log.error("写入数据到starRocks，beanList的长度"+beanList.size());
                Thread.sleep(1000);
            }
        }catch (Exception e){
            e.printStackTrace();
            log.error(e.getMessage());
        }
    }



}
